package org.niit.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming_05 {

    // jion 将两个流 相同的数据 进行合并 （了解）
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.sparkContext.setLogLevel("ERROR")

    //分别监听套接字 两个端口 9999 8848
    val lines9999: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val lines8898: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 8898)


    val map9999: DStream[(String, String)] = lines9999.map((_, "9999"))
    val map8898: DStream[(String, String)] = lines8898.map((_, "8898"))

    val joinDS: DStream[(String, (String, String))] = map9999.join(map8898)

    //Spark Streaming  必须要有结束 否则不能执行  print foreach
//    lines9999.print()
//    lines8898.print()
      joinDS.print()


    ssc.start()
    ssc.awaitTermination()

  }

}
